package io.jnet.server.businessprocessor;

import java.util.concurrent.ExecutorService;
import com.jfireframework.baseutil.concurrent.CpuCachePadingInt;
import com.jfireframework.baseutil.concurrent.MPSCQueue;
import io.jnet.common.streamprocessor.StreamProcessor;
import io.jnet.common.util.ProcesserUtil;
import io.jnet.server.api.ServerChannelContext;
import io.jnet.server.api.ServerListener;
import io.jnet.server.util.ProcessorTask;

public class TaskAttachProcessor implements Runnable
{
    private final MPSCQueue<ProcessorTask> tasks  = new MPSCQueue<>();
    private static final int               IDLE   = 0;
    private static final int               WORK   = 1;
    private final CpuCachePadingInt        status = new CpuCachePadingInt(IDLE);
    private final ExecutorService          executorService;
    private final StreamProcessor[]        processors;
    private final ServerChannelContext     serverChannelContext;
    private final ServerListener           serverListener;
    
    public TaskAttachProcessor(ServerChannelContext serverChannelContext, ExecutorService executorService, ServerListener serverListener)
    {
        this.executorService = executorService;
        this.serverChannelContext = serverChannelContext;
        this.serverListener = serverListener;
        processors = serverChannelContext.processors();
    }
    
    @Override
    public void run()
    {
        do
        {
            ProcessorTask task = tasks.poll();
            if (task == null)
            {
                status.set(IDLE);
                if (tasks.isEmpty() == false)
                {
                    tryExecute();
                }
                return;
            }
            try
            {
                ProcesserUtil.process(serverChannelContext, processors, task.getData(), task.getInitIndex());
            }
            catch (Throwable e)
            {
                serverListener.catchException(e, serverChannelContext);
            }
        } while (true);
    }
    
    public void commit(ProcessorTask task)
    {
        tasks.offer(task);
        tryExecute();
    }
    
    private void tryExecute()
    {
        int now = status.value();
        if (now == IDLE && status.compareAndSwap(IDLE, WORK))
        {
            executorService.execute(this);
        }
    }
}
